package com.itheima.pinda.service;

import com.alibaba.fastjson.JSONObject;
import com.itheima.pinda.entity.LocationEntity;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.io.UnsupportedEncodingException;
import java.nio.charset.Charset;

/**
 * netty业务处理
 *
 * @author llhuang10
 * @date 2021/10/21 10:29
 */
@Slf4j
@ChannelHandler.Sharable
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("ServerHandler.channelRead()");
        ByteBuf buf = (ByteBuf) msg;

        try {
            // 接收报文
            String context = this.getRequestBody(buf);
            log.info("报文内容: {}", context);

            // 解析报文
            String message = this.parseMessage(context);
            if (StringUtils.isBlank(message)) {
                log.info("报文解析失败");
                return;
            }

            // 发送至Kafka队列
            KafkaSender.send(KafkaSender.MSG_TOPIC, message);
        } catch (Exception e) {
            log.error("error: {}", e.getMessage(), e);
        } finally {
            // 使用完ByteBuf之后, 需要主动去释放资源, 否则, 资源一直在内存加载, 容易造成内存泄漏
            ReferenceCountUtil.release(msg);
        }

        if (buf != null) {
            // 把当前的写指针 writerIndex 恢复到之前保存的 markedWriterIndex值
            buf.resetWriterIndex();
        }
    }


    /**
     * 解析报文
     *
     * @param context
     * @return
     */
    private String parseMessage(String context) {
        if (StringUtils.isBlank(context)) {
            log.warn("报文为空");
            return null;
        }
        context = context.trim();
        // 其他格式的报文需要解析后放入MessageEntity实体
        LocationEntity message = JSONObject.parseObject(context, LocationEntity.class);
        if (message == null || StringUtils.isBlank(message.getType()) || StringUtils.isBlank(message.getBusinessId())
                || StringUtils.isBlank(message.getLat()) || StringUtils.isBlank(message.getLat()) || StringUtils.isBlank(message.getId())) {
            log.warn("报文内容异常");
            return null;
        }
        return JSONObject.toJSONString(message);
    }


    /**
     * 解析请求内容
     *
     * @param buf ByteBuf
     * @return ByteBuf -> String
     */
    private String getRequestBody(ByteBuf buf) throws UnsupportedEncodingException {
        // buf.readableBytes() 返回可读字节数  writerIndex - readerIndex
        if (buf.readableBytes() <= 0) {
            return null;
        }
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        return new String(req, Charset.defaultCharset().toString());
    }


    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 写一个空的buf, 并刷新写出区域, 完成后关闭sock channel连接
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
                .addListener(ChannelFutureListener.CLOSE);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 关闭发生异常的连接
        ctx.close();
    }
}
